package com.gollum.mq;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson2.JSON;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.gollum.common.domain.redis.MqListenerMethod;
import com.gollum.common.domain.redis.MqMessage;
import com.gollum.common.utils.redis.RedisCache;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.connection.stream.Consumer;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 消息队列注册
 *
 * @author Herther
 * @createTime 2022年11月02日 23:41:00
 * @since 1.0.0
 */
@Slf4j
public class MqMessageQueueRegister implements ApplicationRunner, ApplicationContextAware {
    //private final Logger logger = LoggerFactory.getLogger(RedisMessageQueueRegister.class);

    private ApplicationContext applicationContext;

    @Autowired
    private RedisCache redisCache;

    @Autowired
    private StreamMessageListenerContainer<String, ObjectRecord<String, String>> streamMessageListenerContainer;

    @Autowired
    private ObjectMapper objectMapper;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        // 启动redis消息队列监听器
        Map<String, List<MqListenerMethod>> candidates = MqListenerAnnotationScan.getCandidates();

        for (Map.Entry<String, List<MqListenerMethod>> entry : candidates.entrySet()) {
            String queueKey = entry.getKey();
            List<MqListenerMethod> redisListenerMethodList = entry.getValue();
            String[] split = queueKey.split("-");
            String streamKey = split[0];
            String consumerGroupName = split[1];
            String consumerName = split[2];
            try {
                List<String> groupName =  redisCache.getStreamGroups(streamKey);
                if (!groupName.contains(consumerGroupName)) {
                    redisCache.createGroup(streamKey,consumerGroupName);
                }
            } catch (Exception e) {
                redisCache.createGroup(streamKey,consumerGroupName);
            }

            streamMessageListenerContainer.receive(
                    Consumer.from(consumerGroupName, consumerName),
                    StreamOffset.create(streamKey, ReadOffset.lastConsumed()),
                    message -> {
                        try {

                            log.info("stream message。messageId={}, stream={}, body={}",
                                    message.getId(), message.getStream(), message.getValue());
                            String messageJson = message.getValue();
                            for (MqListenerMethod rlm : redisListenerMethodList) {
                                Method targetMethod = rlm.getTargetMethod();
                                try {
                                    if (rlm.getMethodParameterClassName().equals(MqMessage.class.getName())) {
                                        messageJson = formatJson(messageJson);
                                        MqMessage<?> redisMessage = objectMapper.readValue(messageJson, MqMessage.class);
                                        targetMethod.invoke(rlm.getBean(applicationContext), redisMessage);
                                    } else {
                                        JavaType javaType = objectMapper.getTypeFactory().constructParametricType(MqMessage.class, rlm.getParameterClass());
                                        messageJson = formatJson(messageJson);
                                        MqMessage<?> redisMessage = objectMapper.readValue(messageJson, javaType);
                                        targetMethod.invoke(rlm.getBean(applicationContext), redisMessage.getData());
                                    }
                                } catch (IllegalAccessException | InvocationTargetException e) {
                                    e.printStackTrace();
                                }
                            }
                            Long ack = redisCache.ack(consumerGroupName, message);
                            log.info("消费确认ack: [{}]",ack);
                        } catch (Exception e) {
                            log.error("消费异常：[{}]", e);
                        }
                    });
            log.info("启动消息队列监听器：[{}.{}]",streamKey,consumerGroupName);
        }
    }



    /**
     * 将转义字符去除
     *
     * @param messageJson json字符串
     * @return String 去除后的
     * @auther: Herther
     * @since 1.0.0
     * @date: 2022/11/3 23:26
     */
    private String formatJson(String messageJson){
        int i = messageJson.indexOf("{");
        messageJson = messageJson.substring(i);
        int p = messageJson.lastIndexOf("}");
        messageJson = messageJson.substring(i-1,p+1);
        messageJson = messageJson.replaceAll("\\\\","");
        return messageJson;
    }
}
